package demo.java.util.concurrent;

import java.time.LocalDateTime;
import java.util.Random;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import demo.java.lang.ThreadDemo;

public class ScheduledExecutorServiceDemo {

    static Logger logger = LoggerFactory.getLogger(ScheduledExecutorServiceDemo.class);

    private static ScheduledExecutorService scheduledThreadPool = new ScheduledThreadPoolExecutor(1,
            new ThreadFactoryDemo.NamedThreadFactory("scheduled"), new AbortPolicy());

    static Random random = new Random();

    /**
     * <h2>任务调度线程池ScheduledThreadPoolExecutor</h2>
     * 先来看看ScheduledThreadPoolExecutor的实现模型，它通过继承ThreadPoolExecutor来重用线程池的功能，里面做了几件事情：
     * 
     * <li>为线程池设置了一个DelayedWorkQueue，该queue同时具有PriorityQueue（优先级大的元素会放到队首）
     * 和DelayQueue（如果队列里第一个元素的getDelay返回值大于0，则take调用会阻塞）的功能
     * 
     * <li>将传入的任务封装成ScheduledFutureTask，这个类有两个特点，实现了java.lang.Comparable和java.util.concurrent.Delayed接口，也就是说里面有两个重要的方法：
     * compareTo和getDelay。 ScheduledFutureTask里面存储了该任务距离下次调度还需要的时间
     * （使用的是基于System#nanoTime实现的相对时间，不会因为系统时间改变而改变，如距离下次执行还有10秒，不会因为将系统时间调前6秒而变成4秒后执行）。
     * getDelay方法就是返回当前时间（运行getDelay的这个时刻）距离下次调用之间的时间差；
     * compareTo用于比较两个任务的优先关系，距离下次调度间隔较短的优先级高。
     * 那么，当有任务丢进上面说到的DelayedWorkQueue时，因为它有DelayQueue（DelayQueue的内部使用PriorityQueue来实现的）的功能，
     * 所以新的任务会与队列中已经存在的任务进行排序，距离下次调度间隔短的任务排在前面，也就是说这个队列并不是先进先出的；
     * 另外，在调用DelayedWorkQueue的take方法的时候，如果没有元素，会阻塞，
     * 如果有元素而第一个元素的getDelay返回值大于0（前面说过已经排好序了，第一个元素的getDelay不会大于后面元素的getDelay返回值），也会一直阻塞。
     * 
     * <li>ScheduledFutureTask提供了一个run的实现，线程池执行的就是这个run方法。
     *
     */
    @Test
    public void scheduledThreadPoolExecutor() {
        ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);
        AtomicLong i = new AtomicLong(0);
//         如果抛出异常，则任务将不再被调度
        exec.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (i.longValue() > 3) {
                    logger.error("抛异常咯，该任务将不再被调度");
                    throw new RuntimeException();
                }
                logger.info("任务1：{}", i.incrementAndGet());
            }
        }, 1000, 200, TimeUnit.MILLISECONDS);

//        每隔一段时间打印系统时间，证明两者是互不影响的 
        exec.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                logger.info("任务2");
            }
        }, 20, 1000, TimeUnit.MILLISECONDS);

        ThreadDemo.safeSleep(10_000);
        exec.shutdown();
    }

    /**
     * 创建一个定期的任务，这个任务将在给定的initial delay后开始执行，并且依次在给定的间隔时间后执行。
     * <li>如果哪个任务执行时候发生了异常，后续就不会在执行任务了。
     * <li>任务只能通过cancel来取消，或者关闭线程池。
     * <li>如果一个任务的执行时间超过了间隔时间，那么下一次任务执行将会被推迟，两个任务不会同时执行。
     */
    @Test
    public void scheduleAtFixedRate() {
        // 最近一次任务开始时间
        AtomicLong latestTaskEndTime = new AtomicLong(0);
        // 最近一次任务结束时间
        AtomicLong latestTaskStartTime = new AtomicLong(0);
//         the period between successive executions
        int period = 2000;
        scheduledThreadPool.scheduleAtFixedRate(() -> {
            long latestTaskStartTimestamp = latestTaskStartTime.longValue();
            if (latestTaskStartTimestamp != 0) {
                logger.info("距离上次任务开始时间：{} ms", System.currentTimeMillis() - latestTaskStartTimestamp);
            }
            latestTaskStartTime.set(System.currentTimeMillis());
            long latestTaskEndTimestamp = latestTaskEndTime.longValue();
            if (latestTaskEndTimestamp != 0) {
                logger.info("距离上次任务结束时间：{} ms", System.currentTimeMillis() - latestTaskEndTimestamp);
            }
            logger.info("task start：{}, period={}", LocalDateTime.now(), period);
            int tmp = random.nextInt(4000);
            ThreadDemo.safeSleep(tmp);
            logger.info("task cost {} ms", tmp);
            latestTaskEndTime.set(System.currentTimeMillis());
        }, 1, period, TimeUnit.MILLISECONDS);
        ThreadDemo.safeSleep(20_000);
        scheduledThreadPool.shutdown();
    }

    /**
     * 创建一个定期执行的任务在initialDelay时间之后开始执行，以后每个任务执行都是在上一个任务完成后的delay时间后才开始执行。
     * <li>如果一个任务执行抛了异常，所有任务都会停止。
     */
    @Test
    public void scheduleWithFixedDelay() {
//        the delay between the termination of one execution and the commencement of the next
        int delay = 2000;
        // 最近一次任务开始时间
        AtomicLong latestTaskEndTime = new AtomicLong(0);
        // 最近一次任务结束时间
        AtomicLong latestTaskStartTime = new AtomicLong(0);
        scheduledThreadPool.scheduleWithFixedDelay(() -> {
            long latestTaskStartTimestamp = latestTaskStartTime.longValue();
            if (latestTaskStartTimestamp != 0) {
                logger.info("距离上次任务开始时间：{} ms", System.currentTimeMillis() - latestTaskStartTimestamp);
            }
            latestTaskStartTime.set(System.currentTimeMillis());
            long latestTaskEndTimestamp = latestTaskEndTime.longValue();
            if (latestTaskEndTimestamp != 0) {
                logger.info("距离上次任务结束时间：{} ms", System.currentTimeMillis() - latestTaskEndTimestamp);
            }
            logger.info("start，delay {} ms, starTime：{}", delay, LocalDateTime.now());
            int tmp = random.nextInt(4000);
            ThreadDemo.safeSleep(tmp);
            logger.info("end,cost {} ms", tmp);
            latestTaskEndTime.set(System.currentTimeMillis());
        }, 1, delay, TimeUnit.MILLISECONDS);
        ThreadDemo.safeSleep(20_000);
        scheduledThreadPool.shutdown();
    }

    /**
     * 一次性任务，延迟N久后执行
     */
    @Test
    public void schedule() {
        ScheduledFuture task = scheduledThreadPool.schedule(() -> {
            logger.info("任务开始：{}", LocalDateTime.now());
            int tmp = random.nextInt(4000);
            ThreadDemo.safeSleep(tmp);
            logger.info("任务执行耗时 {} ms", tmp);
        }, 2000, TimeUnit.MILLISECONDS);
        boolean canceled = task.cancel(false);
        logger.info("cancel task return :{}", canceled);
        ThreadDemo.safeSleep(10_000);
        scheduledThreadPool.shutdown();
    }

    /**
     * 如果抛出异常，则线程WAITING (parking)，该定时任务不会再被调度
     */
    @Test
    public void testScheduledThreadPool() {
        ScheduledExecutorService scheduledThreadPool = new ScheduledThreadPoolExecutor(1,
                new ThreadFactoryDemo.NamedThreadFactory("调度线程池"), new AbortPolicy());
        // 5秒后执行任务
        scheduledThreadPool.schedule(new Runnable() {
            @Override
            public void run() {
                logger.info("schedulePool.schedule");
            }
        }, 5, TimeUnit.SECONDS);

        // 2秒后执行任务，以后每1秒执行一次
        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            int count = 0;

            @Override
            public void run() {
                count++;
                logger.info("schedulePool.scheduleAtFixedRate");
                if (count == 3) {// 抛出异常，线程 WAITING (parking)
                    logger.info("抛异常咯");
                    throw new RuntimeException();
                }
            }
        }, 1, 2, TimeUnit.SECONDS);

        try {
            boolean termination = scheduledThreadPool.awaitTermination(10, TimeUnit.SECONDS);
            logger.info("termination:{}", termination);
        } catch (InterruptedException e) {
            logger.error("", e);
        }
    }

    public void test() {
    }
}
